package com.aididi.config;

import com.aididi.consumer.OrderMessageListener;
import com.aididi.exception.BizException;
import com.aididi.utils.SM4Util;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author ：xuzexiang
 * @description：消费者配置
 * @date ： 2021/10/28 9:59
 */
@Configuration
public class ConsumerConfig {

    @Bean("mqConsumerConfig")
    @ConfigurationProperties(ignoreInvalidFields = true,prefix = "rocketmq.consumer")
    public MQConsumerProperties mqConsumerConfig(){
        return new MQConsumerProperties();
    }

    @Bean(name = "defaultMQPushConsumer")
    public DefaultMQPushConsumer defaultMQPushConsumer(MQConsumerProperties mqConsumerConfig) throws Exception {
        DefaultMQPushConsumer mqPushConsumer  = null;
        if("Y".equalsIgnoreCase(mqConsumerConfig.getAclIsOnOff()) || "true".equalsIgnoreCase(mqConsumerConfig.getAclIsOnOff()) || "on".equalsIgnoreCase(mqConsumerConfig.getAclIsOnOff())){
            //是否开启acl
            // 国产解密方法
             String aclSecretKey = SM4Util.sm4dec("B001",mqConsumerConfig.getAclSecretKey());
            AclClientRPCHook aclClientRPCHook = new AclClientRPCHook(new SessionCredentials(mqConsumerConfig.getAclAccessKey(), aclSecretKey ));
            //AclClientRPCHook aclClientRPCHook = new AclClientRPCHook(new SessionCredentials(mqConsumerConfig.getAclAccessKey(),  mqConsumerConfig.getAclSecretKey()));
            //AllocateMessageQueueAveragely(平均分配策略):消息队列分配策略算法,
            mqPushConsumer  = new DefaultMQPushConsumer(mqConsumerConfig.getGroupName(),aclClientRPCHook,new AllocateMessageQueueAveragely());
        }else {
            mqPushConsumer  = new DefaultMQPushConsumer(mqConsumerConfig.getGroupName());
        }

        mqPushConsumer.setNamesrvAddr(mqConsumerConfig.getNameSrvAddr());
        //注意：由于消费者线程池使用的是无界队列，因此消费者消费线程数最多只有consumeThreadMin个（线程池的最大线程数只有在线程池使用队列是有界队列，并且有界队列已满，并且核心线程都在运行时，线程池才会创建新的线程，直到数量等于最大线程数
        mqPushConsumer.setConsumeThreadMin(mqConsumerConfig.getConsumeThreadMin());
        mqPushConsumer.setConsumeThreadMax(mqConsumerConfig.getConsumeThreadMax());
        //设置消费者第一次启动从队列头部开始消费还是尾部开始消费，如果非第一次启动，那么按照上次消费的位置继续消费
        //CONSUME_FROM_FIRST_OFFSET:第一次启动从队列初始位置消费，后续再启动接着上次消费的进度开始消费
        mqPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //设置消费模式，集群/广播，默认时集群
        //广播消费： 消息会发给Consume Group中的每一个消费者进行消费,如果设置为广播消息会导致NOT_ONLINE异常,https://github.com/apache/rocketmq/issues/296
        mqPushConsumer.setMessageModel(MessageModel.CLUSTERING);
        //设置一次消费消息的条数，默认一条
        mqPushConsumer.setConsumeMessageBatchMaxSize(mqConsumerConfig.getConsumeMessageBatchMaxSize());
        try {
            /**
             * MQ信息接收处理类绑定
             */
            mqPushConsumer.subscribe(mqConsumerConfig.getTopic(),mqConsumerConfig.getTag());
            // 过滤条件，只消费TagFilterTest主题下的TagA 和 TagB下的消息
            //consumer.subscribe("TagFilterTest", "TagA || TagC");
            OrderMessageListener orderMessageListener = new OrderMessageListener(mqConsumerConfig.getTopic(),mqConsumerConfig.getTag(),mqConsumerConfig.getReConsumerTimes());
            //自定义mq消费者消息监听器(推模式),不能保证有序消费
            mqPushConsumer.registerMessageListener(orderMessageListener);
            //消费端顺序消费
            /*mqPushConsumer.registerMessageListener(new MessageListenerOrderly() {
                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                    consumeOrderlyContext.setAutoCommit(true);
                    for(MessageExt msg:list){
                        System.out.println("收到消息内容 "+new String(msg.getBody()));
                    }
                    return ConsumeOrderlyStatus.SUCCESS;
                }
            });*/
            mqPushConsumer.start();
        }catch (MQClientException e){
            throw new BizException(e);
        }

        return mqPushConsumer;

    }


    }
